package org.bear.bundle.framework;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQPrefetchPolicy;

/**
 * JMS消息消费者
 * 
 */
public class JMSConsumer implements ExceptionListener {

	// 队列预取策略
	private int queuePrefetch = DEFAULT_QUEUE_PREFETCH;
	public final static int DEFAULT_QUEUE_PREFETCH = 1;

	private String brokerUrl;

	private String username;

	private String password;

	private MessageListener messageListener;

	private Connection connection;

	private Session session;
	// 队列名
	private String queue;

	public JMSConsumer(String brokerUrl, String username, String password,
			String ip) {
		this(brokerUrl, username, password, ip, DEFAULT_QUEUE_PREFETCH);
	}

	public JMSConsumer(String brokerUrl, String username, String password,
			String ip, int queuePrefetch) {
		this.brokerUrl = brokerUrl;
		this.username = username;
		this.password = password;
		this.queuePrefetch = queuePrefetch;
	}

	/**
	 * 执行消息获取的操作
	 * 
	 * @throws Exception
	 */
	public void start() throws Exception {
		// ActiveMQ的连接工厂
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
				this.username, this.password, this.brokerUrl);
		connection = connectionFactory.createConnection();
		// activeMQ预取策略
		ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy();
		prefetchPolicy.setQueuePrefetch(queuePrefetch);
		((ActiveMQConnection) connection).setPrefetchPolicy(prefetchPolicy);
		connection.setExceptionListener(this);
		connection.start();
		// 会话采用非事务级别，消息到达机制使用自动通知机制
		// AUTO_ACKNOWLEDGE ：
		// 指定消息提供者在每次收到消息时自动发送确认。消息只向目标发送一次，但传输过程中可能因为错误而丢失消息。
		// CLIENT_ACKNOWLEDGE ：
		// 由消息接收者确认收到消息，通过调用消息的acknowledge()方法（会通知消息提供者收到了消息）
		// DUPS_OK_ACKNOWLEDGE ：
		// 指定消息提供者在消息接收者没有确认发送时重新发送消息（这种确认模式不在乎接收者收到重复的消息）。
		session = connection.createSession(Boolean.FALSE,
				Session.AUTO_ACKNOWLEDGE);
		Destination destination = session.createQueue(this.queue);
		MessageConsumer consumer = session.createConsumer(destination);
		consumer.setMessageListener(this.messageListener);
	}

	public MessageConsumer getMessageConsumer(String queueName)
			throws Exception {
		// ActiveMQ的连接工厂
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
				this.username, this.password, this.brokerUrl);
		connection = connectionFactory.createConnection();
		// activeMQ预取策略
		ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy();
		prefetchPolicy.setQueuePrefetch(queuePrefetch);
		((ActiveMQConnection) connection).setPrefetchPolicy(prefetchPolicy);
		connection.setExceptionListener(this);
		connection.start();
		// 会话采用非事务级别，消息到达机制使用：会通知消息提供者收到了消息
		session = connection.createSession(Boolean.FALSE,
				Session.AUTO_ACKNOWLEDGE);
		Destination destination = session.createQueue(queueName);
		MessageConsumer consumer = session.createConsumer(destination);

		return consumer;
	}

	/**
	 * 关闭连接
	 */
	public void shutdown() {
		try {
			if (session != null) {
				session.close();
				session = null;
			}
			if (connection != null) {
				connection.close();
				connection = null;
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			Runtime.getRuntime().gc();
		}
	}

	@Override
	public void onException(JMSException e) {
		e.printStackTrace();
	}

	public void setMessageListener(MessageListener messageListener, String queue) {
		this.messageListener = messageListener;
		this.queue = queue;
	}

}
